using System;
using System.Runtime.CompilerServices;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Diagnostics;
using MQTTnet.Extensions.Rpc;
using MQTTnet.Formatter;
using MQTTnet.Protocol;

namespace HslCommunicationEx.Mqtt
{
    public class MqttNetClient
    {
        private readonly MqttFactory _mqttFactory = new MqttFactory();

        private Task _taskReconnect;

        public IMqttClient MqtClient { get; set; }

        public IMqttNetLogger ClientLogger { get; } = new MqttNetLogger();


        public bool ClearSession { get; set; }

        public int KeepAliveSeconds { get; set; }

        public bool IsConnected => this.MqtClient?.IsConnected ?? false;

        public int ReconnectInterval { get; set; } = 2000;


        public Action<MqttClientConnectedEventArgs> ActOnConnectedHandler { get; set; }

        public Action<MqttClientDisconnectedEventArgs> ActOnDisConnectedHandler { get; set; }

        public Action<MqttApplicationMessageReceivedEventArgs> ActOnMessageReceivedHandler { get; set; }

        public Action<Exception> ErrorHandler { get; set; }

        public bool AllowReconnect { get; set; } = true;


        public async Task<bool> Connect(string server, int port, string userName, string pwd, string clientId, int timeout = 3000, MqttClientTypeEnum mqttClientType = MqttClientTypeEnum.Tcp, MqttProtocolVersion version = MqttProtocolVersion.V500)
        {
            IMqttClient client = await this.Connect2(server, port, userName, pwd, clientId, timeout, mqttClientType, version);
            if (client != null)
            {
                this.MqtClient = client;
            }
            return client?.IsConnected ?? false;
        }

        public async Task<IMqttClient> Connect2(string server, int port, string userName, string pwd, string clientId, int timeout = 3000, MqttClientTypeEnum mqttClientType = MqttClientTypeEnum.Tcp, MqttProtocolVersion version = MqttProtocolVersion.V500)
        {
            try
            {
                MqttClientOptions options = new MqttClientOptions
                {
                    ClientId = clientId,
                    ProtocolVersion = version,
                    Timeout = TimeSpan.FromMilliseconds(timeout),
                    KeepAlivePeriod = TimeSpan.FromSeconds(this.KeepAliveSeconds),
                    CleanSession = this.ClearSession
                };
                options.Credentials = new MqttClientCredentials(userName, Encoding.UTF8.GetBytes(pwd));
                MqttClientTlsOptions tlsOptions = new MqttClientTlsOptions
                {
                    UseTls = false,
                    IgnoreCertificateChainErrors = true,
                    IgnoreCertificateRevocationErrors = true,
                    AllowUntrustedCertificates = true
                };
                switch (mqttClientType)
                {
                    case MqttClientTypeEnum.Tcp:
                        options.ChannelOptions = new MqttClientTcpOptions
                        {
                            Server = server,
                            Port = port,
                            TlsOptions = tlsOptions
                        };
                        break;
                    case MqttClientTypeEnum.Wss:
                        {
                            new MqttClientOptions().ClientId = clientId;
                            MqttClientWebSocketOptions mqttClientWebSocketOptions = new MqttClientWebSocketOptions();
                            string url = string.Format("{0}:{1}/mqtt", server, port);
                            //		DefaultInterpolatedStringHandler defaultInterpolatedStringHandler = new DefaultInterpolatedStringHandler(6, 2);
                            //defaultInterpolatedStringHandler.AppendFormatted(server);
                            //defaultInterpolatedStringHandler.AppendLiteral(":");
                            //defaultInterpolatedStringHandler.AppendFormatted(port);
                            //defaultInterpolatedStringHandler.AppendLiteral("/mqtt");
                            //mqttClientWebSocketOptions.Uri = defaultInterpolatedStringHandler.ToStringAndClear();
                            mqttClientWebSocketOptions.Uri = url;
                            mqttClientWebSocketOptions.TlsOptions = tlsOptions;
                            options.ChannelOptions = mqttClientWebSocketOptions;
                            break;
                        }
                }
                if (this.MqtClient != null)
                {
                    await this.MqtClient.DisconnectAsync();
                }
                IMqttClient client = (this.MqtClient = this._mqttFactory.CreateMqttClient(this.ClientLogger));
                client.ConnectedAsync += OnConnectedHandler;
                client.DisconnectedAsync += Reconnect;
                client.ApplicationMessageReceivedAsync += OnMessageReceivedHandler;
                await client.ConnectAsync(options);
                return client;
            }
            catch (Exception)
            {
                return null;
            }
        }

        private Task Reconnect(MqttClientDisconnectedEventArgs e)
        {
            try
            {
                this.ActOnDisConnectedHandler?.Invoke(e);
            }
            catch (Exception)
            {
            }
            if (this.AllowReconnect && this._taskReconnect == null)
            {
                this._taskReconnect = Task.Run(async delegate
                {
                    while (true)
                    {
                        try
                        {
                            IMqttClient mqtClient = this.MqtClient;
                            if (mqtClient == null || !mqtClient.IsConnected)
                            {
                                await this.MqtClient.ConnectAsync(this.MqtClient.Options);
                                if (this.MqtClient?.IsConnected ?? false)
                                {
                                    this._taskReconnect = null;
                                    return;
                                }
                            }
                        }
                        catch (Exception)
                        {
                        }
                        Thread.Sleep(this.ReconnectInterval);
                    }
                });
            }
            return Task.CompletedTask;
        }

        public async Task SendManualPing()
        {
            await this.MqtClient.PingAsync(CancellationToken.None);
        }

        public async Task ReconnectAsync()
        {
            if (!(this.MqtClient?.IsConnected ?? false))
            {
                await this.MqtClient.ConnectAsync(this.MqtClient.Options);
            }
        }

        public async Task DisconnectAsync()
        {
            if (this.MqtClient != null)
            {
                await this.MqtClient.DisconnectAsync(new MqttClientDisconnectOptions());
            }
        }

        public async Task<bool> RpcSubMethod(IMqttClient client, string methodName, Func<MqttApplicationMessageReceivedEventArgs, byte[]> rcvHandler)
        {
            IMqttClient mqttClient = client;
            if (mqttClient == null || !mqttClient.IsConnected)
            {
                return false;
            }
            await client.SubscribeAsync("MQTTnet.RPC/+/" + methodName);
            client.ApplicationMessageReceivedAsync += async delegate (MqttApplicationMessageReceivedEventArgs e)
            {
                byte[] sendBuf = rcvHandler?.Invoke(e);
                MqttApplicationMessage message = new MqttApplicationMessageBuilder().WithTopic(e.ApplicationMessage.Topic + "/response").WithPayload(sendBuf).Build();
                if (sendBuf != null)
                {
                    await client.PublishAsync(message);
                }
            };
            return true;
        }

        public async Task<byte[]> RpcPubExcute(IMqttClient client, string methodName, byte[] sendBuf, int timeout = 30000)
        {
            MqttFactory mqttFactory = new MqttFactory();
            using IMqttRpcClient mqttRpcClient = mqttFactory.CreateMqttRpcClient(client);
            return await mqttRpcClient.ExecuteAsync(TimeSpan.FromSeconds(timeout), methodName, sendBuf, MqttQualityOfServiceLevel.AtMostOnce);
        }

        public async Task<byte[]> RpcPubExcute(string server, int port, string methodName, byte[] sendBuf, int timeout = 30000)
        {
            MqttFactory mqttFactory = new MqttFactory();
            using IMqttClient mqttClient = mqttFactory.CreateMqttClient();
            MqttClientOptions mqttClientOptions = new MqttClientOptionsBuilder().WithTcpServer(server, port).Build();
            await mqttClient.ConnectAsync(mqttClientOptions);
            using IMqttRpcClient mqttRpcClient = mqttFactory.CreateMqttRpcClient(mqttClient);
            return await mqttRpcClient.ExecuteAsync(TimeSpan.FromSeconds(timeout), methodName, sendBuf, MqttQualityOfServiceLevel.AtMostOnce);
        }

        private Task OnConnectedHandler(MqttClientConnectedEventArgs e)
        {
            try
            {
                this.ActOnConnectedHandler?.Invoke(e);
            }
            catch (Exception obj)
            {
                this.ErrorHandler?.Invoke(obj);
            }
            return Task.CompletedTask;
        }

        private Task OnMessageReceivedHandler(MqttApplicationMessageReceivedEventArgs e)
        {
            try
            {
                this.ActOnMessageReceivedHandler?.Invoke(e);
            }
            catch (Exception obj)
            {
                this.ErrorHandler?.Invoke(obj);
            }
            return Task.CompletedTask;
        }

        public OperateRet SubTopic(string topic)
        {
            OperateRet operateRet = new OperateRet();
            try
            {
                MqttClientSubscribeResult result = this.MqtClient.SubscribeAsync(topic).Result;
                operateRet.Message = result.ReasonString;
                operateRet.Success = true;
                return operateRet;
            }
            catch (Exception ex)
            {
                operateRet.Message = ex.Message;
                return operateRet;
            }
        }

        public OperateRet PublishSync(string topic, byte[] data, MqttQualityOfServiceLevel level = MqttQualityOfServiceLevel.AtMostOnce, bool retainFlag = false)
        {
            OperateRet operateRet = new OperateRet();
            try
            {
                MqttApplicationMessage applicationMessage = new MqttApplicationMessageBuilder().WithTopic(topic).WithPayload(data).WithQualityOfServiceLevel(level)
                    .WithRetainFlag(retainFlag)
                    .Build();
                MqttClientPublishResult result = this.MqtClient.PublishAsync(applicationMessage).Result;
                operateRet.Success = result.ReasonCode == MqttClientPublishReasonCode.Success;
                if (!operateRet.Success)
                {
                    operateRet.Message = result.ReasonString;
                }
                return operateRet;
            }
            catch (Exception ex)
            {
                operateRet.Message = ex.ToString();
                return operateRet;
            }
        }
    }
}
